#include "external_sorter.h"  // 引入 ExternalSorter 的类声明与依赖

#include <algorithm>          // std::sort
#include <fstream>            // 文件读写流
#include <iomanip>            // std::setw 等格式化输出
#include <iostream>           // 标准输出
#include <random>             // 随机数生成器
#include <sstream>            // std::ostringstream
#include <thread>             // std::thread

namespace fs = std::filesystem; // 方便使用 filesystem 命名空间

ExternalSorter::ExternalSorter(Config cfg)
    : cfg_(std::move(cfg)) {}  // 使用成员初始化列表，接管配置对象

void ExternalSorter::run() {
    if (fs::exists(cfg_.temp_dir)) {        // 如果临时目录已经存在
        fs::remove_all(cfg_.temp_dir);      //    先清理，保证干净环境
    }

    generate_input();   // 阶段 0：生成输入数据
    launch_pipeline();  // 阶段 1~3：读/排/写流水线
    merge_runs();       // 阶段 4：多路归并 run
    cleanup();          // 阶段 5：清理临时目录
}

void ExternalSorter::generate_input() {
    std::ofstream out(cfg_.input_file);      // 打开（或创建）输入文件
    if (!out) {                              // 若文件打开失败
        throw std::runtime_error("failed to open input file for writing"); // 抛异常
    }

    std::mt19937_64 rng{cfg_.rng_seed};      // 64 位 Mersenne Twister 随机数引擎
    std::uniform_int_distribution<std::int64_t> dist(-5'000'000, 5'000'000); // 均匀分布

    std::cout << "[generate] writing " << cfg_.total_numbers
              << " numbers to " << cfg_.input_file << '\n'; // 打印生成信息

    for (std::size_t i = 0; i < cfg_.total_numbers; ++i) {   // 生成 total_numbers 个随机数
        out << dist(rng) << '\n';                            // 每个数独占一行写入文件
    }
}

void ExternalSorter::launch_pipeline() {
    std::thread reader(                         // 启动读线程
        reader_thread,                          // 线程入口函数
        std::cref(cfg_.input_file),             // 输入文件路径（常量引用包装）
        cfg_.chunk_size,                        // 每块大小
        std::ref(read_to_sort_));               // 输出队列的引用

    std::thread sorter(                         // 启动排序线程
        sorter_thread,
        std::ref(read_to_sort_),                // 输入队列（来自读线程）
        std::ref(sort_to_write_));              // 输出队列（给写线程）

    std::thread writer(                         // 启动写线程
        writer_thread,
        std::cref(cfg_.temp_dir),               // run 输出目录
        std::ref(sort_to_write_),               // 输入队列（来自排序线程）
        std::ref(writer_ctx_));                 // 写线程上下文（记录文件路径）

    reader.join();  // 等待读线程结束，防止主线程过早退出
    sorter.join();  // 等待排序线程结束
    writer.join();  // 等待写线程结束
}

void ExternalSorter::merge_runs() {
    std::vector<std::string> run_paths;        // 归并阶段要用的 run 路径列表
    {
        std::lock_guard<std::mutex> lock(writer_ctx_.mutex); // 上锁保护 paths
        for (const auto& path : writer_ctx_.paths) {          // 遍历记录的路径
            if (!path.empty()) {                             // 忽略未写入的位置
                run_paths.push_back(path);                    // 收集有效路径
            }
        }
    }

    std::cout << "[main] total sorted runs: " << run_paths.size() << '\n'; // 打印 run 数量

    multiway_merge_runs(run_paths, cfg_.output_file); // 执行多路归并

    if (verify_sorted_file(cfg_.output_file)) {       // 校验最终文件是否升序
        std::cout << "[verify] output is sorted ascendingly.\n";
    } else {
        throw std::runtime_error("[verify] output NOT sorted!"); // 发现逆序立即报错
    }
}

void ExternalSorter::cleanup() {
    fs::remove_all(cfg_.temp_dir);                   // 删除临时目录（及其中的 run）
    std::cout << "[cleanup] temporary directory removed.\n";
}

// =============================== 静态工具函数 ===============================

void ExternalSorter::reader_thread(const std::string& input_path,
                                   std::size_t chunk_size,
                                   BlockingQueue<Chunk>& queue) {
    std::ifstream in(input_path);                      // 打开输入文件
    if (!in) {
        throw std::runtime_error("failed to open input file for reading");
    }

    std::size_t chunk_id = 0;                          // 记录块编号
    while (true) {
        Chunk chunk;                                   // 创建新的块对象
        chunk.id = chunk_id++;                         // 赋予唯一编号
        chunk.data.reserve(chunk_size);                // 预分配容量

        std::int64_t value;
        for (std::size_t i = 0; i < chunk_size && (in >> value); ++i) {
            chunk.data.push_back(value);               // 逐个读入数据
        }

        if (chunk.data.empty()) {                      // 如果没有读到数据
            break;                                     // 退出循环
        }

        queue.push(std::move(chunk));                  // 将块推入排序队列
    }

    queue.close();                                     // 通知排序线程：数据已全部推送
    std::cout << "[reader] finished, total chunks: " << chunk_id - 1 << '\n';
}

void ExternalSorter::sorter_thread(BlockingQueue<Chunk>& in_queue,
                                   BlockingQueue<Chunk>& out_queue) {
    Chunk chunk;                                       // 用于接收队列中的块
    std::size_t sorted = 0;                            // 计数：共排序多少块
    while (in_queue.pop(chunk)) {                      // 一直取到队列关闭且为空为止
        std::sort(chunk.data.begin(), chunk.data.end()); // 在内存中排序
        out_queue.push(std::move(chunk));              // 将排序好的块传给写线程
        ++sorted;
    }
    out_queue.close();                                 // 通知写线程：不会再有新块
    std::cout << "[sorter] chunks sorted: " << sorted << '\n';
}

void ExternalSorter::writer_thread(const fs::path& output_dir,
                                   BlockingQueue<Chunk>& queue,
                                   WriterContext& ctx) {
    if (!fs::exists(output_dir)) {                     // 若 run 目录不存在
        fs::create_directories(output_dir);            // 创建目录层级
    }

    Chunk chunk;                                       // 接收排序后的块
    std::size_t written = 0;                           // 记录写出的块数
    while (queue.pop(chunk)) {                         // 循环直到队列关闭且无数据
        std::ostringstream name;
        name << "chunk_"                                // 文件名前缀
             << std::setw(4) << std::setfill('0')       // 使用 4 位宽度，左侧补零
             << chunk.id << ".txt";
        fs::path chunk_path = output_dir / name.str(); // 构造 run 文件完整路径

        std::ofstream out(chunk_path);                 // 打开 run 文件
        if (!out) {
            throw std::runtime_error("failed to open chunk file for writing");
        }
        for (std::int64_t value : chunk.data) {        // 顺序写出块内数据
            out << value << '\n';
        }

        {
            std::lock_guard<std::mutex> lock(ctx.mutex); // 保护 paths
            if (chunk.id >= ctx.paths.size()) {          // 如果数组长度不够
                ctx.paths.resize(chunk.id + 1);           //    扩容到包含当前 id
            }
            ctx.paths[chunk.id] = chunk_path.string();   // 记录 run 文件路径
        }

        ++written;
    }
    std::cout << "[writer] chunks written: " << written << '\n';
}

void ExternalSorter::multiway_merge_runs(const std::vector<std::string>& run_paths,
                                         const std::string& output_path) {
    struct Node {
        std::int64_t value;    // 当前值
        std::size_t run_index; // 来源 run 的下标

        bool operator<(const Node& rhs) const {
            return value > rhs.value; // 反向比较：在 max-heap 上模拟 min-heap
        }
    };

    std::vector<std::ifstream> runs;               // 持有所有 run 的输入流
    runs.reserve(run_paths.size());
    std::vector<Node> heap;                        // 最小堆
    heap.reserve(run_paths.size());

    for (std::size_t i = 0; i < run_paths.size(); ++i) {
        runs.emplace_back(run_paths[i]);           // 打开第 i 个 run
        if (!runs.back()) {
            throw std::runtime_error("failed to open run file: " + run_paths[i]);
        }
        std::int64_t value;
        if (runs.back() >> value) {                // 读取 run 的第一条记录
            heap.push_back({value, i});            // 加入堆中
        }
    }

    std::ofstream out(output_path);                // 打开最终输出文件
    if (!out) {
        throw std::runtime_error("failed to open output file: " + output_path);
    }

    std::make_heap(heap.begin(), heap.end());      // 建堆（默认 max-heap + 自定义比较函数）
    std::size_t emitted = 0;                       // 累计写出的记录数

    while (!heap.empty()) {                        // 堆为空说明全部 run 耗尽
        std::pop_heap(heap.begin(), heap.end());   // 将最小元素移到末尾
        Node node = heap.back();                   // 取出堆尾（当前最小值）
        heap.pop_back();

        out << node.value << '\n';                 // 写入输出文件
        ++emitted;

        std::int64_t next_value;
        if (runs[node.run_index] >> next_value) {  // 从同一 run 读取下一个值
            heap.push_back({next_value, node.run_index}); // 仍有数据则继续参与归并
            std::push_heap(heap.begin(), heap.end());     // 调整堆结构
        }
    }

    std::cout << "[merge] total records emitted: " << emitted << '\n';
}

bool ExternalSorter::verify_sorted_file(const std::string& path) {
    std::ifstream in(path);                          // 打开输出文件
    if (!in) {
        throw std::runtime_error("failed to open file for verification: " + path);
    }

    std::int64_t prev, current;
    if (!(in >> prev)) {                             // 若文件为空
        return true;                                 //    直接判定为已排序
    }
    while (in >> current) {                          // 逐行读取后续数据
        if (current < prev) {                        // 若发现逆序
            return false;                            //    验证失败
        }
        prev = current;                              // 更新上一元素
    }
    return true;                                     // 整个文件保持非降序
}